package com.bruce.starts.demo.thread.forkjoin;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public class WordCount {
    public static void main(String[] args) {
        String[] fc = {"hello world",
                "hello me",
                "hello fork",
                "hello join",
                "fork join in world"};
        // 创建 ForkJoin 线程池
        ForkJoinPool fjp = new ForkJoinPool(3);
        MR mr = new MR(fc,0,fc.length);
        Map<String,Long> result = fjp.invoke(mr);
        result.forEach((k,v)-> System.out.println(k+":"+v));
    }
    static class MR extends RecursiveTask<Map<String, Long>> {
        private String[] fc;
        private int start, end;

        public MR(String[] fc, int start, int end) {
            this.fc = fc;
            this.start = start;
            this.end = end;
        }

        @Override
        protected Map<String, Long> compute() {
            if (end - start == 1) {
                return cacl(fc[start]);
            }else {
                int mid = (end+start)/2;
                MR mr1 = new MR(fc,start,mid);
                mr1.fork();
                MR mr2 = new MR(fc,mid,end);
                return merge(mr2.compute(),mr1.join());
            }
        }

        private Map<String, Long> merge(Map<String, Long> map1, Map<String, Long> map2) {
            Map<String, Long> result = new HashMap<>();
            result.putAll(map1);
            // 合并结果
            map2.forEach((k, v) -> {
                Long c = result.get(k);
                if (c != null) {
                    result.put(k, c + v);
                } else {
                    result.put(k, v);
                }
            });
            return result;
        }

        private Map<String, Long> cacl(String line) {
            Map<String, Long> result = new HashMap<>();
            String[] words = line.split("\\s+");
            Arrays.stream(words).forEach(w -> {
                Long v = result.get(w);
                if (v != null) {
                    result.put(w, v + 1);
                } else {
                    result.put(w, 1L);
                }
            });
            return result;
        }
    }
}
